package com.htiiot.hbase;
import java.io.IOException;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.hbase.HConstants;
import org.apache.hadoop.hbase.classification.InterfaceAudience;
import org.apache.hadoop.hbase.classification.InterfaceStability;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.filter.CompareFilter;
import org.apache.hadoop.hbase.filter.Filter;
import org.apache.hadoop.hbase.filter.PrefixFilter;
import org.apache.hadoop.hbase.filter.RegexStringComparator;
import org.apache.hadoop.hbase.filter.RowFilter;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.mapreduce.HFileOutputFormat2;
import org.apache.hadoop.hbase.mapreduce.TableInputFormat;
import org.apache.hadoop.hbase.mapreduce.TableMapReduceUtil;
import org.apache.hadoop.hbase.mapreduce.TableMapper;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.GenericOptionsParser;

@InterfaceAudience.Public
@InterfaceStability.Stable
public class GHfile {
    private static final Log LOG =
            LogFactory.getLog(GHfile.class.getName());


    static final String NAME = "GH";

    static class CellCounterMapper
            extends TableMapper<ImmutableBytesWritable, Cell> {

        @Override
        public void map(ImmutableBytesWritable row, Result values,
                        Context context)
                throws IOException, InterruptedException {

                for (Cell value : values.listCells()) {
                    context.write(row,value);
                }
        }
    }

    public static Job createSubmittableJob(Configuration conf, String[] args)
            throws IOException {
        String tableName = args[0];
        String reportSeparatorString = (args.length > 2) ? args[2]: ":";
        conf.set("ReportSeparator", reportSeparatorString);
        Job job = new Job(conf, NAME + "_" + tableName);
        job.setJarByClass(GHfile.class);
        Scan scan = getConfiguredScanForJob(conf, args);
        TableMapReduceUtil.initTableMapperJob(tableName, scan,
                CellCounterMapper.class, ImmutableBytesWritable.class, Cell.class, job);
        job.setNumReduceTasks(0);
        job.setMapOutputKeyClass(ImmutableBytesWritable.class);
        job.setMapOutputValueClass(Cell.class);
        job.setOutputFormatClass(HFileOutputFormat2.class);
        FileOutputFormat.setOutputPath(job,new Path(args[1]));
        return job;
    }

    private static Scan getConfiguredScanForJob(Configuration conf, String[] args) throws IOException {
        Scan s = new Scan();
        // Set Scan Versions
        s.setMaxVersions(Integer.MAX_VALUE);
        s.setCacheBlocks(false);
        // Set Scan Column Family
        if (conf.get(TableInputFormat.SCAN_COLUMN_FAMILY) != null) {
            s.addFamily(Bytes.toBytes(conf.get(TableInputFormat.SCAN_COLUMN_FAMILY)));
        }
        // Set RowFilter or Prefix Filter if applicable.
        Filter rowFilter = getRowFilter(args);
        if (rowFilter!= null) {
            LOG.info("Setting Row Filter for counter.");
            s.setFilter(rowFilter);
        }
        // Set TimeRange if defined
        long timeRange[] = getTimeRange(args);
        if (timeRange != null) {
            LOG.info("Setting TimeRange for counter.");
            s.setTimeRange(timeRange[0], timeRange[1]);
        }
        return s;
    }


    private static Filter getRowFilter(String[] args) {
        Filter rowFilter = null;
        String filterCriteria = (args.length > 3) ? args[3]: null;
        if (filterCriteria == null) return null;
        if (filterCriteria.startsWith("^")) {
            String regexPattern = filterCriteria.substring(1, filterCriteria.length());
            rowFilter = new RowFilter(CompareFilter.CompareOp.EQUAL, new RegexStringComparator(regexPattern));
        } else {
            rowFilter = new PrefixFilter(Bytes.toBytes(filterCriteria));
        }
        return rowFilter;
    }

    private static long[] getTimeRange(String[] args) throws IOException {
        final String startTimeArgKey = "--starttime=";
        final String endTimeArgKey = "--endtime=";
        long startTime = 0L;
        long endTime = 0L;

        for (int i = 1; i < args.length; i++) {
            System.out.println("i:" + i + "arg[i]" + args[i]);
            if (args[i].startsWith(startTimeArgKey)) {
                startTime = Long.parseLong(args[i].substring(startTimeArgKey.length()));
            }
            if (args[i].startsWith(endTimeArgKey)) {
                endTime = Long.parseLong(args[i].substring(endTimeArgKey.length()));
            }
        }

        if (startTime == 0 && endTime == 0)
            return null;

        endTime = endTime == 0 ? HConstants.LATEST_TIMESTAMP : endTime;
        return new long [] {startTime, endTime};
    }
    /**
     * Main entry point.
     *
     * @param args The command line parameters.
     * @throws Exception When running the job fails.
     */
    public static void main(String[] args) throws Exception {
        Configuration conf = HBaseConfiguration.create();
        String[] otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs();
        if (otherArgs.length < 2) {
            System.err.println("ERROR: Wrong number of parameters: " + args.length);
            System.err.println("Usage: CellCounter ");
            System.err.println("       <tablename> <outputDir> <reportSeparator> [^[regex pattern] or " +
                    "[Prefix] for row filter]] --starttime=[starttime] --endtime=[endtime]");
            System.err.println("  Note: -D properties will be applied to the conf used. ");
            System.err.println("  Additionally, the following SCAN properties can be specified");
            System.err.println("  to get fine grained control on what is counted..");
            System.err.println("   -D " + TableInputFormat.SCAN_COLUMN_FAMILY + "=<familyName>");
            System.err.println(" <reportSeparator> parameter can be used to override the default report separator " +
                    "string : used to separate the rowId/column family name and qualifier name.");
            System.err.println(" [^[regex pattern] or [Prefix] parameter can be used to limit the cell counter count " +
                    "operation to a limited subset of rows from the table based on regex or prefix pattern.");
            System.exit(-1);
        }
        Job job = createSubmittableJob(conf, otherArgs);
        System.exit(job.waitForCompletion(true) ? 0 : 1);
    }
}
